use crate::func::{ExecutorFunc, RedisCache};
use crate::report::get_col_name_exps;
use crate::{get_group_by, set_leaf_columns_props, Column, ColumnLevel, GroupCalc};
use clickhouse_rs::ClientHandle;
use redis::Commands;
use shq_common::prelude::clickhouse::{get_columns, table_is_exist};
use shq_common::prelude::report::{ReportGroup, Unit};
use shq_common::prelude::{
    get_ch_pool, get_named_lock_manager, get_redis_cli, parse_exp, Align, DataType, Parameter,
};
use std::borrow::Cow;
use std::sync::Arc;
use tokio::sync::Mutex;

use super::{create_leaf_cell_column, get_days_exp, get_leaf_columns, HKEY_CROSS_TABLE_COLS};

//按照执行的参数，执行sql语句
pub(crate) struct SqlExecutor<'a> {
    rept_name: &'a str,
    params: &'a [(&'a str, &'a str)],
    tmp_table_name: String,
}
impl<'a> SqlExecutor<'a> {
    pub(crate) fn new(rept_name: &'a str, params: &'a [(&'a str, &'a str)]) -> Self {
        let mut tmp_table_name = format!("tmp_{}_", rept_name);
        for (_k, v) in params.iter() {
            tmp_table_name.push_str(v);
            tmp_table_name.push('_');
        }
        tmp_table_name.pop();
        SqlExecutor {
            rept_name,
            params,
            tmp_table_name,
        }
    }
    // 获取基于报表名称和参数的锁
    async fn get_lock(&self) -> Arc<Mutex<()>> {
        let lock_manager = get_named_lock_manager().await;
        lock_manager.get_lock(self.tmp_table_name.as_str()).await
    }
    pub(crate) async fn exec(
        &self,
        sql_str: &str,
        leaf_cols: &[&Column],
        conf_params: &[Parameter],
        groups: &[ReportGroup],
        unique_columns: &[String],
    ) -> anyhow::Result<&str> {
        let tname = self.tmp_table_name.as_str();
        let exist = table_is_exist(tname).await?;
        if exist {
            return Ok(tname);
        }
        let lock = self.get_lock().await;
        let _guard = lock.lock().await;
        let exist = table_is_exist(tname).await?;
        if exist {
            return Ok(tname);
        }
        let mut client = get_ch_pool().get_handle().await?;
        let sql_str = parse_exp(sql_str, self.params.as_ref())?;
        let mut sqls: Vec<&str> = sql_str.split(';').collect();
        let mut slen = sqls.len();
        if slen > 0 && sqls[slen - 1].trim().is_empty() {
            sqls.pop();
            slen -= 1;
        };
        for (i, sql) in sqls.iter().enumerate() {
            let sql = sql.trim();
            if sql.is_empty() {
                continue;
            }
            if i == slen - 1 {
                let group_bys = get_group_by(groups, None);
                let col_name_exps =
                    get_col_name_exps(&leaf_cols, group_bys.as_slice(), conf_params, self.params);
                let days_exp = get_days_exp(self.params, conf_params);
                let with_exp = if let Some(days_exp) = days_exp {
                    Some(format!("with {days_exp}"))
                } else {
                    None
                };
                self.exec_last_sql(
                    &mut client,
                    sql,
                    leaf_cols,
                    &col_name_exps,
                    groups,
                    unique_columns,
                    group_bys.as_slice(),
                    with_exp.as_deref(),
                )
                .await?;
            } else {
                client.execute(sql).await?;
            }
        }
        Ok(tname)
    }
    async fn delte_tmp_table(&self) -> anyhow::Result<()> {
        let tname = self.tmp_table_name.as_str();
        let mut client = get_ch_pool().get_handle().await?;
        let sql_str = format!("drop table if exists {}", tname);
        client.execute(&sql_str).await?;
        Ok(())
    }
    async fn exec_last_sql(
        &self,
        client: &mut ClientHandle,
        sql: &str,
        leaf_cols: &[&Column],
        col_name_exps: &[(&str, Cow<'_, str>, bool)],
        groups: &[ReportGroup],
        unique_columns: &[String],
        group_bys: &[&str],
        with_exp: Option<&str>,
    ) -> anyhow::Result<()> {
        let select_str = col_name_exps
            .iter()
            .map(|(k, v, sn)| format!("{v} as {}{k}", if *sn == true { "_" } else { "" }))
            .collect::<Vec<String>>()
            .join(",");
        let group_str = if !group_bys.is_empty() {
            format!("group by {}", group_bys.join(","))
        } else {
            "".to_string()
        };
        let table_sql = if sql.contains(|c| c == ' ' || c == '\n' || c == '\r') {
            format!("({sql})")
        } else {
            sql.to_string()
        };
        let glen = groups.len();
        let gidx = if glen > 0 { glen - 1 } else { 0 };
        let with_exp = with_exp.unwrap_or("");
        let comma = if select_str.is_empty() { "" } else { "," };
        let unique_cols_str = unique_columns
            .iter()
            .map(|c| c.as_str())
            .collect::<Vec<&str>>()
            .join(",");
        let sql_str = format!(
            "create table if not exists {} Engine=MergeTree ORDER BY({unique_cols_str}) as {with_exp} select {}{comma}toUInt8({gidx}) gp_index from {table_sql} {group_str}",
            self.tmp_table_name, select_str
        );
        client.execute(&sql_str).await?;
        let rename_sqls: Vec<String> = col_name_exps
            .iter()
            .filter(|e| e.2)
            .map(|(k, _, _)| {
                format!(
                    "alter table {} RENAME COLUMN _{k} to {k}",
                    self.tmp_table_name
                )
            })
            .collect();
        for rename_sql in rename_sqls.into_iter() {
            client.execute(rename_sql).await?;
        }
        if !groups.is_empty() {
            let glen = groups.len();
            for i in 0..glen - 1 {
                self.create_group(leaf_cols, groups, i).await?;
            }
        }
        Ok(())
    }
    pub(crate) async fn exec_cross(
        &self,
        sql_str: &str,
        columns: &[Column],
        rows: &[Column],
        cells: &[Column],
        col_levs: &[ColumnLevel],
        groups: &[ReportGroup],
        unique_columns: &[String],
        conf_params: &[Parameter],
        unit: Unit,
    ) -> anyhow::Result<&str> {
        let tname = self.tmp_table_name.as_str();
        let exist = table_is_exist(&tname).await?;
        let mut client = get_ch_pool().get_handle().await?;
        if exist {
            return Ok(tname);
        }
        let lock = self.get_lock().await;
        let _guard = lock.lock().await;
        let exist = table_is_exist(tname).await?;
        if exist {
            return Ok(tname);
        }
        let sql_str = parse_exp(sql_str, self.params.as_ref())?;
        let (mut cols, with_exp) = self
            .create_cross_columns(
                &mut client,
                &sql_str,
                columns,
                rows,
                cells,
                col_levs,
                conf_params,
            )
            .await?;
        let leafs = get_leaf_columns(&cols);
        let group_bys = get_group_by(groups, Some(rows));
        let col_name_exps =
            get_col_name_exps(&leafs, group_bys.as_slice(), conf_params, self.params);
        let with_exp = if !with_exp.is_empty() {
            Some(with_exp.as_str())
        } else {
            None
        };
        self.exec_last_sql(
            &mut client,
            &sql_str,
            &leafs,
            &col_name_exps,
            groups,
            unique_columns,
            group_bys.as_slice(),
            with_exp,
        )
        .await?;
        let mut unique_cols = groups.iter().map(|g| g.by.as_str()).collect::<Vec<&str>>();
        for c in col_levs.iter() {
            unique_cols.push(c.id_col.as_str());
        }
        for c in rows.iter() {
            if c.data_type == DataType::Int {
                unique_cols.push(c.name.as_deref().unwrap_or_default());
            }
        }
        let col_types = self.get_columns().await?;
        set_leaf_columns_props(&mut cols, &unique_cols, &col_types, unit)?;
        let cols_str = serde_yaml::to_string(&cols)?;
        let mut rcli = get_redis_cli().get_connection()?;
        let _: () = rcli.hset(HKEY_CROSS_TABLE_COLS, tname, cols_str)?;
        Ok(tname)
    }
    fn get_column(
        &self,
        id_val: &str,
        name_val: &str,
        pid: &str,
        cells: &[Column],
        items: &Vec<(String, String, Option<String>)>,
        col_levs: &[ColumnLevel],
        sub_total_calc: Option<&GroupCalc>,
    ) -> Column {
        let mut children: Vec<Column> = Vec::new();
        if sub_total_calc.is_none() {
            children = items
                .iter()
                .filter(|(_, _, pid)| pid.as_deref() == Some(id_val))
                .map(|(id, name, pid)| {
                    self.get_column(
                        id.as_str(),
                        name.as_str(),
                        pid.as_deref().unwrap_or_default(),
                        cells,
                        items,
                        col_levs,
                        None,
                    )
                })
                .collect();
            if children.len() > 1 {
                let idx = if pid == "" { 0 } else { pid.split('_').count() };
                let cl = col_levs.get(idx).unwrap();
                if let Some(sub_total_calc) = cl.sub_total_calc.as_ref() {
                    let subtotal = self.get_column(
                        id_val,
                        name_val,
                        pid,
                        cells,
                        items,
                        col_levs,
                        Some(sub_total_calc),
                    );
                    children.push(subtotal);
                }
            }
        }
        let col = if children.len() == 0 || sub_total_calc.is_some() {
            if cells.len() > 1 {
                let children = cells
                    .iter()
                    .map(|cell| {
                        create_leaf_cell_column(cell, id_val, pid, col_levs, sub_total_calc)
                    })
                    .collect();
                let title = if let Some(sub_total_calc) = sub_total_calc {
                    format!("{}{}", name_val, sub_total_calc.to_str())
                } else {
                    name_val.to_owned()
                };
                Column::new(
                    None,
                    Some(title),
                    None,
                    GroupCalc::Any,
                    Align::Left,
                    false,
                    None,
                    children,
                    None,
                )
            } else {
                create_leaf_cell_column(&cells[0], id_val, pid, col_levs, sub_total_calc)
            }
        } else {
            Column::new(
                None,
                Some(name_val.to_owned()),
                None,
                GroupCalc::Any,
                Align::Left,
                false,
                None,
                children,
                None,
            )
        };
        col
    }
    fn get_total_columns(&self, cells: &[Column]) -> Vec<Column> {
        let mut columns: Vec<Column> = Vec::new();
        if cells.len() > 1 {
            let mut cols: Vec<Column> = Vec::new();
            for cell in cells.iter() {
                let col = create_leaf_cell_column(cell, "", "", &[], None);
                cols.push(col);
            }
            if !cols.is_empty() {
                let col = Column::new(
                    None,
                    Some("汇总".to_string()),
                    None,
                    GroupCalc::default(),
                    Align::default(),
                    false,
                    None,
                    cols,
                    None,
                );
                columns.push(col);
            }
        } else {
            create_leaf_cell_column(&cells[0], "", "", &[], None);
        }
        columns
    }
    async fn create_cross_columns(
        &self,
        client: &mut ClientHandle,
        sql_str: &str,
        columns: &[Column],
        rows: &[Column],
        cells: &[Column],
        col_levs: &[ColumnLevel],
        conf_params: &[Parameter],
    ) -> anyhow::Result<(Vec<Column>, String)> {
        let mut sel_cols: Vec<String> = Vec::new();
        let mut ord_cols: Vec<&str> = Vec::new();
        let mut with_cols: Vec<String> = Vec::new();
        let days_exp = get_days_exp(self.params, conf_params);
        if let Some(days_exp) = days_exp {
            with_cols.push(days_exp)
        }
        col_levs.iter().for_each(|cl| {
            let id_col_exp = columns
                .iter()
                .find(|c| c.name.as_deref().unwrap_or_default() == cl.id_col)
                .unwrap()
                .col_exp
                .as_deref();
            if let Some(id_col_exp) = id_col_exp {
                with_cols.push(format!("{} as {}", id_col_exp, cl.id_col));
            }
            let name_col_exp = columns
                .iter()
                .find(|c| c.name.as_deref().unwrap_or_default() == cl.name_col)
                .unwrap()
                .col_exp
                .as_deref();
            if let Some(name_col_exp) = name_col_exp {
                with_cols.push(format!("{} as {}", name_col_exp, cl.name_col));
            }
            sel_cols.push(format!("toString({})", cl.id_col));
            sel_cols.push(format!("toString({})", cl.name_col));
            if let Some(oc) = &cl.order_col {
                ord_cols.push(oc);
            } else {
                ord_cols.push(&cl.id_col);
            }
        });
        let sel_cols_str = sel_cols.join(",");
        let with_cols_str = if !with_cols.is_empty() {
            format!("with {}", with_cols.join(","))
        } else {
            "".to_string()
        };
        let sql_str = format!(
            "{with_cols_str} select distinct {sel_cols_str} from ({sql_str}) order by {}",
            ord_cols.join(",")
        );
        let block = client.query(sql_str).fetch_all().await?;
        let mut items: Vec<(String, String, Option<String>)> = Vec::new();
        let clen = col_levs.len();
        for row in block.rows() {
            let mut p_id: Option<String> = None;
            for i in 0..clen {
                let id: String = row.get(i * 2)?;
                let exist = items.iter().any(|(id2, _, _)| id2 == &id);
                if !exist {
                    let name: String = row.get(i * 2 + 1)?;
                    items.push((id.clone(), name, p_id.clone()));
                }
                if i > 0 && i < clen {
                    p_id = if let Some(pid) = p_id {
                        Some(format!("{}_{}", pid, id))
                    } else {
                        Some(id.clone())
                    };
                };
            }
        }
        if items.is_empty() {
            return Err(anyhow::anyhow!("没有查询到数据"));
        }
        let mut columns: Vec<Column> = Vec::new();
        for row in rows.iter() {
            columns.push(row.to_owned().to_owned());
        }
        let top_id_names: Vec<(&str, &str, &str)> = items
            .iter()
            .filter(|(_, _, id)| *id == None)
            .map(|(id, name, pid)| {
                (
                    id.as_str(),
                    name.as_str(),
                    pid.as_deref().unwrap_or_default(),
                )
            })
            .collect();
        for (id, name, pid) in top_id_names {
            let col = self.get_column(id, name, pid, cells, &items, col_levs, None);
            columns.push(col);
        }
        self.get_total_columns(cells).iter().for_each(|c| {
            columns.push(c.to_owned());
        });
        Ok((columns, with_cols_str))
    }
    pub(crate) async fn exec_func(
        &self,
        func_name: &str,
        leaf_cols: &[&Column],
        groups: &[ReportGroup],
        rept_unit: &Unit,
    ) -> anyhow::Result<&str> {
        let tname = self.tmp_table_name.as_str();
        let mut rc = RedisCache::new(tname)?;
        let exist = rc.is_exist()? || table_is_exist(tname).await?;
        if exist {
            return Ok(tname);
        }
        let lock = self.get_lock().await;
        let _guard = lock.lock().await;
        let exist = rc.is_exist()? || table_is_exist(tname).await?;
        if exist {
            return Ok(tname);
        }
        ExecutorFunc::new(func_name)
            .exec(
                self.rept_name,
                self.params,
                leaf_cols,
                rept_unit,
                self.tmp_table_name.as_str(),
            )
            .await?;
        if !groups.is_empty() {
            let glen = groups.len();
            for i in 0..glen - 1 {
                self.create_group(leaf_cols, groups, i).await?;
            }
        }
        Ok(self.tmp_table_name.as_str())
    }
    async fn create_group(
        &self,
        leaf_cols: &[&Column],
        groups: &[ReportGroup],
        gidx: usize,
    ) -> anyhow::Result<()> {
        let by_cols: Vec<&str> = groups
            .iter()
            .enumerate()
            .filter(|(i, _)| *i <= gidx)
            .map(|(_, g)| g.by.as_str())
            .collect();
        let by_col_names: Vec<&[String]> = groups
            .iter()
            .enumerate()
            .filter(|(i, _)| *i <= gidx)
            .map(|(_, g)| g.name_cols.as_slice())
            .collect();
        let other_by_cols: Vec<(&str, &[String])> = groups
            .iter()
            .enumerate()
            .filter(|(i, _)| *i > gidx)
            .map(|(_, g)| (g.by.as_str(), g.name_cols.as_slice()))
            .collect();
        let subtotal = groups.get(gidx).and_then(|g| {
            let name_col = groups.get(gidx + 1).and_then(|g| {
                if gidx < groups.len() - 1 {
                    Some(g.name_cols.get(0).map(|s| s.as_str()).unwrap_or_default())
                } else {
                    None
                }
            });
            if let Some(name_col) = name_col {
                if gidx < groups.len() - 1 {
                    if let Some(subtotal) = g.subtotal.as_ref() {
                        let cname = g.name_cols.get(0).map(|s| s.as_str()).unwrap_or_default();
                        Some((subtotal, cname, name_col))
                    } else {
                        None
                    }
                } else {
                    None
                }
            } else {
                None
            }
        });
        let select_str = leaf_cols
            .iter()
            .map(|c| {
                let col_name = c.name.as_deref().unwrap();
                if by_cols.contains(&col_name) {
                    col_name.to_string()
                } else if subtotal.map(|st| st.2 == col_name).unwrap_or(false) {
                    let (st, name_val_col, name_col) = subtotal.unwrap();
                    format!("concat({name_val_col}, '{}') as {}", st.suffix, name_col)
                } else if by_col_names
                    .iter()
                    .any(|names| names.iter().any(|name| name == col_name))
                {
                    format!("anyLast({col_name}) as {}", col_name)
                } else if other_by_cols.iter().any(|c| c.0 == col_name) {
                    format!("0 as {}", col_name)
                } else if other_by_cols
                    .iter()
                    .any(|c| c.1.iter().any(|s| s == col_name))
                {
                    format!("'' as {}", col_name)
                } else {
                    let has_gp_col = c
                        .col_exp
                        .as_ref()
                        .map(|exp| exp.contains("gp_index="))
                        .unwrap_or(false);
                    let (is_subtotal_col,agg_exp) = if let Some((st, _, _)) = subtotal {
                        (st.agg_cols.iter().any(|ac| ac == col_name),
                        st.agg_exps.get(col_name))
                    } else {
                        (true, None)
                    };
                    if has_gp_col {
                        let exp = c.col_exp.as_deref().unwrap_or("");
                        format!("{exp} as {col_name}")
                    } else if is_subtotal_col {
                        if let Some(agg_exp) = agg_exp {
                            return format!("{agg_exp} as {col_name}");
                        }
                        let func = match c.data_type {
                            DataType::Int => {
                                if by_cols.contains(&col_name) {
                                    "anyLast"
                                } else {
                                    "sum"
                                }
                            }
                            DataType::Float => "sum",
                            _ => "anyLast",
                        };
                        format!("{func}({col_name}) as {col_name}")
                    } else {
                        match c.data_type {
                            DataType::Bool => format!("false as {}", col_name),
                            DataType::Int => format!("0 as {}", col_name),
                            DataType::Float => format!("0.0 as {}", col_name),
                            DataType::String => format!("'' as {}", col_name),
                            DataType::Date => format!("'1970-01-01' as {}", col_name),
                            DataType::Datetime => format!("'1970-01-01 00:00:00' as {}", col_name),
                        }
                    }
                }
            })
            .collect::<Vec<String>>()
            .join(",");
        let mut client = get_ch_pool().get_handle().await?;
        let group_by_exp = by_cols.join(",");
        let tname = self.tmp_table_name.as_str();
        let max_gidx = groups.len() - 1;
        let sql_str = format!(
            "insert into {tname} select {select_str},toUInt8({gidx}) gp_index from {tname} where {tname}.gp_index = {max_gidx} group by {group_by_exp}"
        );
        client.execute(&sql_str).await?;
        Ok(())
    }
    pub(crate) async fn get_columns(&self) -> anyhow::Result<Vec<(String, DataType)>> {
        let mut rc = RedisCache::new(self.tmp_table_name.as_str())?;
        if rc.is_exist()? {
            return rc.get_columns();
        }
        get_columns(self.tmp_table_name.as_str())
            .await?
            .into_iter()
            .map(|(k, v)| Ok((k, v.into())))
            .collect()
    }
}
